package com.wuwangfu.source;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.UUID;

/**
 * @Author jcshen
 * @Date 2023-02-21
 * @PackageName:com.wuwangfu.source
 * @ClassName: CustomSourceR
 * @Description: TODO
 * @Version 1.0
 * 在web ui直接 cancel job
 * <p>
 * http://localhost:8081
 */
public class CustomSourceR {
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
        env.setParallelism(2);

        DataStreamSource<String> line = env.addSource(new MyRSource());
        System.out.println(line.getParallelism());

        line.print();
        env.execute();
    }

    /**
     * 继承RichParallelSourceFunction抽象类的Source，是多并行度
     */
    private static class MyRSource extends RichParallelSourceFunction<String> {

        /**
         * 1、调用 MyRSource 构造方法
         * 2、调用open方法，调用一次
         * 3、调用run方法，
         * 4、调用cancel方法停止run方法的循环
         * 5、调用close方法，释放资源
         */

        private boolean flag = true;
        int indexSubTask;

        public MyRSource() {
            System.out.println("constructor invoked");
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            indexSubTask = getRuntimeContext().getIndexOfThisSubtask();
            System.out.println("subTask-" + indexSubTask + " open method invoked");
        }

        @Override
        public void close() throws Exception {
            System.out.println("subTask-" + indexSubTask + " close method invoked");
        }

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            System.out.println("subTask-" + indexSubTask + " run method invoked");

            while (flag) {
                ctx.collect("subTask-" + indexSubTask + ": " + UUID.randomUUID().toString());
                Thread.sleep(2000);
            }
        }

        @Override
        public void cancel() {
            System.out.println("subTask-" + indexSubTask + " cancel method invoked");
            flag = false;
        }
    }
}
